package itemCF.step2;


import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.text.DecimalFormat;
import java.util.ArrayList;
import java.util.List;

/**
 * @author legolas
 * @date 2020/3/13 下午7:06

 */

public class Mapper2 extends Mapper<LongWritable, Text, Text, Text> {

    private List<String> cacheList = new ArrayList<>();
    private DecimalFormat df = new DecimalFormat("0.00");

    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        super.setup(context);
        try {
            //步骤1的输出评分矩阵
            FileReader fr = new FileReader("itemUserScore2");
            BufferedReader br = new BufferedReader(fr);
            String line = null;
            while ((line = br.readLine()) != null) {
                cacheList.add(line);
            }
            fr.close();
            br.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 需要实现的map函数
     * 这个map函数就是可以接收k1，v1   产生k2，v2
     *
     * @param k1
     * @param v1
     * @param context
     * @throws IOException
     * @throws InterruptedException
     */


    /**
     * 求出余弦相似度：
     * 1    2.0     0       5.0
     * 2    10.0    3.0     0
     * <p>
     * cos<1,2>= 2.0*10.0 + 0*3.0 + 5.0*0 / √2.0^2+0^2+5.0^2 * √10.0^2+3.0^2+0^2
     */

    @Override
    protected void map(LongWritable k1, Text v1, Context context) throws IOException, InterruptedException {
        try {
            //行:物品id
            String row_matix1 = v1.toString().split("\t")[0];
            //列_值数组:用户id_操作权重总分
            String[] column_value_array_matrix1 = v1.toString().split("\t")[1].split(",");
            double denominator1 = 0;
            //计算左侧矩阵行的空间距离
            for (String column_value : column_value_array_matrix1) {
                String score = column_value.split("_")[1];
                denominator1 += Double.valueOf(score) * Double.valueOf(score);
            }
            denominator1 = Math.sqrt(denominator1);

            for (String line : cacheList) {
                //右侧矩阵的行
                //格式：行 tab 列_值，列_值，列_值,列_值
                String row_matrix2 = line.split("\t")[0];
                String[] column_value_array_matrix2 = line.split("\t")[1].split(",");
                double denominator2 = 0;
                for (String colum_value : column_value_array_matrix2) {
                    String score = colum_value.split("_")[1];
                    denominator2 += Double.valueOf(score) * Double.valueOf(score);
                }
                denominator2 = Math.sqrt(denominator2);

                //矩阵两行同列相乘后累加得到的结果
                int numerator = 0;
                //遍历左矩阵的第一行的每一列
                for (String column_value_matrix1 : column_value_array_matrix1) {
                    //物品id
                    String column_matrix1 = column_value_matrix1.split("_")[0];
                    //用户id_分值
                    String value_matrix1 = column_value_matrix1.split("_")[1];

                    //遍历右矩阵每一行的每一列
                    for (String column_value_matrix2 : column_value_array_matrix2) {
                        //将相同用户id的列值相乘，然后累加
                        if (column_value_matrix2.startsWith(column_matrix1 + "_")) {
                            String value_matrix2 = column_value_matrix2.split("_")[1];
                            //将两列的值相乘，并累加
                            numerator += Integer.parseInt(value_matrix1) * Integer.parseInt(value_matrix2);
                        }
                    }
                }
                double cos = numerator / (denominator1 * denominator2);
                if (cos == 0) {
                    continue;
                }
                //result是结果矩阵中的某元素，坐标为 行:row_matrix1，列:row_matrix2
                Text k2 = new Text();
                Text v2 = new Text();
                k2.set(row_matix1);
                v2.set(row_matrix2 + "_" + df.format(cos));
                /**
                 * 最终输出：reduce中做聚合
                 * 1    1_cos<1,1>
                 * 1    2_cos<1,2>
                 * 1    3_cos<1,3>
                 * 2    1_cos<2,1>
                 * 2    2_cos<2,2>
                 */
                context.write(k2, v2);

            }
        } catch (Exception e) {
            e.printStackTrace();
        }


    }
}
